package com.tjh.hive_data_compare;

import com.tjh.hive_data_compare.util.FileUtil;
import com.tjh.hive_data_compare.util.HiveUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


public class StartCompare {
    private static final Logger logger = LogManager.getLogger(StartCompare.class);

    public static void main(String[] args) {
        //用法：第一个参数为源头集群，第二个参数为目标集群hive地址，第三个参数为样例数据库
        //第四个参数为需要对比的数据库，第五个参数为需要对比的表名
        try {
            String oneAddress = "192.168.83.48";
            String twoAddress = "192.168.83.176";
            String sampleDB = "dw_ods";
            //需要对比的库名
            String compareDb = "";
            //需要对比的表多个用|隔开
            String compareTable = "";
            if (args != null && args.length == 2) {
                oneAddress = args[0];
                twoAddress = args[1];
            } else if (args != null && args.length == 3) {
                oneAddress = args[0];
                twoAddress = args[1];
                sampleDB = args[2];
            } else if (args != null && args.length == 4) {
                oneAddress = args[0];
                twoAddress = args[1];
                sampleDB = args[2];
                compareDb = args[3];
            } else if (args != null && args.length == 5) {
                oneAddress = args[0];
                twoAddress = args[1];
                sampleDB = args[2];
                compareDb = args[3];
                compareTable = args[4];
            }

            SimpleDateFormat sdf = new SimpleDateFormat();// 格式化时间
            sdf.applyPattern("yyyy-MM-dd HHmmss");// a为am/pm的标记
            Date date = new Date();// 获取当前时间
            String currenDate = sdf.format(date);
            currenDate = currenDate.replace("\\:", "").replaceAll(" ", "");
            Map<String, Integer> tmpMap1 = getTableNumber(oneAddress, "", "", sampleDB, compareDb, compareTable);
            System.out.println(oneAddress + "获取完成");
            Map<String, Integer> tmpMap2 = getTableNumber(twoAddress, "", "", sampleDB, compareDb, compareTable);
            if (tmpMap1 != null) {
                for (String tmpKey : tmpMap1.keySet()) {
                    int oneResult = tmpMap1.get(tmpKey);
                    int twoResult = -1;
                    if (tmpMap2.containsKey(tmpKey)) {
                        twoResult = tmpMap2.get(tmpKey);
                    } else {
                        logger.error("新集群缺失表:" + tmpKey);
                    }

                    if (oneResult != twoResult) {
                        logger.error("条数不一致:" + tmpKey + " " + oneResult + "|" + twoResult);
                        FileUtil.appendMethodB(currenDate + "_compare.txt", tmpKey + " " + oneResult + "|" + twoResult + "\n");
                        System.out.println(tmpKey + " " + oneResult + "|" + twoResult);
                    } else {
                        FileUtil.appendMethodB(currenDate + "_compare_sample.txt", tmpKey + " " + oneResult + "|" + twoResult + "\n");
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static Map<String, Integer> getTableNumber(String serverUrl, String userName, String passWord, String dbName, String compareDB, String compareTable) {
        Map<String, Integer> tmpMap1 = new HashMap<>();
        try {
            HiveUtil hiveUtil_1 = new HiveUtil();
            hiveUtil_1.init(serverUrl, userName, passWord, dbName);
            List<String> dbList = hiveUtil_1.showDatabases();
            hiveUtil_1.destory();
            if (dbList != null) {
                for (String tmpdb : dbList) {
                    try {
                        if (StringUtils.isNotEmpty(compareDB) && !compareDB.contains(tmpdb)) {
                            continue;
                        }
                        System.out.println("开始处理：" + tmpdb);
                        HiveUtil hiveUtil_2 = new HiveUtil();
                        hiveUtil_2.init(serverUrl, userName, passWord, tmpdb);
                        List<String> tableList = hiveUtil_2.showTables();
                        System.out.println(tableList.size());
                        if (tableList != null) {
                            for (String tmpTable : tableList) {
                                try {
                                    if (StringUtils.isNotEmpty(compareTable) && !compareTable.contains(tmpTable)) {
                                        continue;
                                    }
                                    int tmpResult = hiveUtil_2.count(tmpdb + "." + tmpTable);
                                    tmpMap1.put(tmpdb + "." + tmpTable, tmpResult);
                                } catch (Exception ex) {
                                    logger.error(tmpdb + "." + tmpTable + "有问题:" + ex.getMessage());
                                }
                            }
                        }
                        hiveUtil_2.destory();
                    } catch (Exception ex) {
                        logger.error(tmpdb + "有问题:" + ex.getMessage());
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

        return tmpMap1;
    }
}
